package com.zc58s.springbootmqtt.mqtt;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zc58s.springbootmqtt.mqtt.distributor.Distributor;
import com.zc58s.springbootmqtt.mqtt.vo.SubcribeRes;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import sun.misc.Unsafe;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author : fujc-dev@qq.com
 * @created : 2022/9/22 10:00
 */

@Component
@Slf4j
public class MqttBundle {


    @Value("${kggs.mqtt.host}")
    private String host;
    @Value("${kggs.mqtt.username}")
    private String username;
    @Value("${kggs.mqtt.password}")
    private String password;
    @Value("${kggs.mqtt.client-id}")
    private String clientId;

    @Autowired
    private Distributor distributor;

    public void initMqtt() {
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            client = new MqttClient(host, clientId, persistence);
            client.setCallback(new MQTTCallback());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private static volatile MqttClient client = null;

    public void connect() {
        try {
            if (client == null) {
                this.initMqtt();
            }
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            options.setConnectionTimeout(2);
            options.setKeepAliveInterval(10);
            client.connect(options);
            System.out.println("MQTT连接：" + client.isConnected());
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }

    private synchronized boolean isConnected() {
        return client != null && client.isConnected();
    }

    public void publish(String topic, String msg) {
        try {
            if (!isConnected()) {
                connect();
            }
            MqttMessage message = new MqttMessage();
            message.setQos(Webtalk.qos);
            message.setRetained(true);
            try {
                message.setPayload(msg.getBytes("UTF-8"));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            MqttTopic mqttTopic = client.getTopic(topic);
            MqttDeliveryToken token = mqttTopic.publish(message);
            if (!token.isComplete()) {
                token.waitForCompletion();
            }
        } catch (MqttPersistenceException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅消息）
     *
     * @param topic
     */
    public void subscribe() {
        try {
            if (!isConnected()) {
                connect();
            }
            String[] topics = {Webtalk.WSN_CSN_sub_data_res,
                    Webtalk.WSN_CSN_unsub_data_res,
                    Webtalk.WSN_CSN_pub_data_req,
                    Webtalk.WSN_CSN_repub_data_req,
                    Webtalk.WSN_CSN_pub_hisdata_req,
                    Webtalk.WSN_CSN_pub_cmd_res};
            int[] qos = new int[topics.length];
            for (int i = 0; i < topics.length; i++) {
                qos[i] = Webtalk.qos;
            }
            client.subscribe(topics, qos);
            log.info("消息订阅成功, topic:" + String.join(",", topics));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public class MQTTCallback implements org.eclipse.paho.client.mqttv3.MqttCallback {

        public void connectionLost(Throwable cause) {
            System.out.println("连接断开，可以做重连，" + cause.getMessage());
            subscribe();
        }

        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.println("应答消息内容: 广播完成，" + token.isComplete());
        }

        public void messageArrived(String topic, MqttMessage message) throws Exception {
            System.out.println("===============================================================================================================================================================================");
            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + new String(message.getPayload()));
            ObjectMapper mapper = new ObjectMapper();
            SubcribeRes res = mapper.readValue(new String(message.getPayload()), SubcribeRes.class);
            res.setTopic(topic);
            res.setQos(message.getQos());
            res.setMessageId(message.getId());
            distributor.submit(res);
        }
    }

}
